/*
 * Copyright [2011] [Ambud Sharma/ODep Monitor]

   Licensed under the Apache License, Version 2.0 (the "License");

   you may not use this file except in compliance with the License.

   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software

   distributed under the License is distributed on an "AS IS" BASIS,

   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

   See the License for the specific language governing permissions and

   limitations under the License.
 */

package edu.uwsp.orion.odepm.main.server;


import java.io.IOException;
import java.lang.reflect.Array;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.InterfaceAddress;
import java.net.NetworkInterface;
import java.net.SocketException;
import java.rmi.RMISecurityManager;
import java.rmi.RemoteException;
import java.rmi.registry.LocateRegistry;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import odepnamenodermi.NameRMI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.hdfs.server.namenode.DatanodeDescriptor;
import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;

import com.google.gwt.thirdparty.guava.common.collect.Multiset.Entry;
import com.gwtext.client.widgets.DataView.Data;

import edu.uwsp.orion.odepm.main.shared.NamenodeInfo;
import edu.uwsp.orion.odepm.main.shared.NodeMonitorInfo;
import edu.uwsp.orion.odepm.main.shared.RemoteTrackerInfo;
import edu.uwsp.orion.odepm.main.shared.TrackerSystemStats;



public class HadoopUtils {

	private static int START_WAIT_TIME=1000;
	private static int NAMENODE_UPDATE_TIME=60*1000;
	private static int DATANODE_UPDATE_TIME=60*1000;
	public static NamenodeInfo nameNode;
	public static JobClient jobClient;
	public static List<DatanodeDescriptor> datanaodesLive,datanaodesDead,datanaodesDecom;
	private static NameRMI rmi;	
	
	public static HashMap<String, TrackerSystemStats> allLiveNodes=new HashMap<String, TrackerSystemStats>();
	
	public static int cpuCount=0,totalMemory=0;
	
	public static float CPU=0,memory=0;
	
	static Configuration con;
	
	static String localIP;
	
	static {
			try {
	            Enumeration<NetworkInterface> interfaces=NetworkInterface.getNetworkInterfaces();
	            while(interfaces.hasMoreElements()){
	                NetworkInterface iface=interfaces.nextElement();
	                if(!iface.isLoopback()){
	                    String addressString="";
	                    List<InterfaceAddress> allIfaceAddresses=iface.getInterfaceAddresses();
	                    for(InterfaceAddress ifaceAddress:allIfaceAddresses){
	                        InetAddress address=ifaceAddress.getAddress();
	                        if(!address.isLoopbackAddress()){
	                            addressString=address.getHostAddress();
	                        }
	                    }
	                    localIP=addressString;
	                }
	            }
	        } catch (SocketException ex) {
	            ex.printStackTrace();
	        }   
			con=new Configuration();
	        InetSocketAddress addr=new InetSocketAddress(localIP, 16010);        
			try {
				rmi = (NameRMI) RPC.waitForProxy(NameRMI.class, NameRMI.versionID, addr, con);
				nameNode=rmi.getNameNode();
			} catch (IOException e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			}
	        try{
	         System.out.print(nameNode.node_count);
	        }
	        catch(Exception e){
	            e.printStackTrace();
	        }
		
	        try {
				datanaodesLive=Arrays.asList(rmi.getAllLiveNodes());
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
	        try {
				datanaodesDead=Arrays.asList(rmi.getAllDeadNodes());
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
	        System.out.print("initialized utils\n");
	       
			Thread thUpdaterNamenode=new Thread(){
				@Override
				public void run() {
					while(true)
					{						
						updateNamenode();
						try {
							Thread.sleep(START_WAIT_TIME);
						} catch (InterruptedException e) {							
							e.printStackTrace();
						}
					}
				}
			};
			thUpdaterNamenode.start();
		
			Thread thUpdaterDataNodes=new Thread(){
				@Override
				public void run() {			
					while(true){
						try{
							updateDataNodeUpdater();
						}
						catch(Exception e){
							e.printStackTrace();
						}
						try {
							Thread.sleep(DATANODE_UPDATE_TIME);
						} catch (InterruptedException e) {
							// TODO Auto-generated catch block
							e.printStackTrace();
						}
					}
				}
			};
			thUpdaterDataNodes.start();
			
	        Thread thUpdateAllNode1s=new Thread(){
				@Override
				public void run() {
					while(true){
						try{							
							updateAllNodes();
						}
						catch(Exception e){
							e.printStackTrace();
						}
						try {
							Thread.sleep(15000);
						} catch (InterruptedException e) {						
							e.printStackTrace();
						}
					}
				}
			};
			thUpdateAllNode1s.start();
	}
	
	private static void updateNamenode(){
		try{
			nameNode=rmi.getNameNode();
		}
		catch(Exception e){
			e.printStackTrace();
		}
		try{			
			cpuCount=0;
			totalMemory=0;
			for(DatanodeDescriptor tem:datanaodesLive){				
				String ip=tem.getHost();
				Configuration con=new Configuration();
				InetSocketAddress addr=new InetSocketAddress(InetAddress.getByName(ip).getHostAddress(), 16000);
		        ip=addr.getAddress().getHostAddress();	        
		        if(ip.startsWith("127") || ip.startsWith("local")){
		    	   addr=new InetSocketAddress(localIP, 16000);
		        }		    
		        
			    RemoteTrackerInfo remote= (RemoteTrackerInfo) RPC.waitForProxy(RemoteTrackerInfo.class, RemoteTrackerInfo.versionID, addr, con);
			    cpuCount+=remote.getTrackerInfo().coreCount;
			 //   System.out.println(cpuCount);
			    totalMemory+=remote.getTrackerInfo().memorySize;
			}
			//System.out.println(totalMemory);
		}
		catch(Exception e){
			e.printStackTrace();
		}
	}
	
	private static void updateDataNodeUpdater(){
		try{
			datanaodesLive=Arrays.asList(rmi.getAllLiveNodes());
			datanaodesDead=Arrays.asList(rmi.getAllDeadNodes());
		}
		catch(Exception e){
			e.printStackTrace();
		}
	}
	
	public HadoopUtils() {
		
	}
	

	
	private static void updateAllNodes(){
		ExecutorService threadPool=Executors.newCachedThreadPool();		
		for(DatanodeDescriptor desc:datanaodesLive){
			final DatanodeDescriptor tem=desc;
			GlobalCPUMemoryUpdater temp=new GlobalCPUMemoryUpdater(tem);
			threadPool.execute(temp);
		}		
		threadPool.shutdown();
		if(threadPool.isShutdown()){			
			float avgCPU=0,avgMemory=0;
			for(Map.Entry<String,TrackerSystemStats> enter:allLiveNodes.entrySet()){
				avgCPU+=enter.getValue().currentCPUUsage;
				avgMemory+=enter.getValue().currentMemory;
			}
			try{
				CPU=(float)avgCPU*100/cpuCount;				
				memory=(float)avgMemory*100/totalMemory;				
			}
			catch(Exception e){
				e.printStackTrace();
			}
		}
	}
	
	private synchronized static void addHashEntry(String key, TrackerSystemStats stat){
		allLiveNodes.put(key, stat);
	}
	
	public static class GlobalCPUMemoryUpdater implements Runnable{

		DatanodeDescriptor tem;
		
		public GlobalCPUMemoryUpdater(DatanodeDescriptor desc) {
			tem=desc;
		}
		
		@Override
		public void run() {
			try{
				String ip=tem.getHost();
				Configuration con=new Configuration();
				InetSocketAddress addr=new InetSocketAddress(InetAddress.getByName(ip).getHostAddress(), 16000);
		        ip=addr.getAddress().getHostAddress();	        
		        if(ip.startsWith("127") || ip.startsWith("local")){
		    	   addr=new InetSocketAddress(localIP, 16000);
		        }		    
		        
			    RemoteTrackerInfo remote= (RemoteTrackerInfo) RPC.waitForProxy(RemoteTrackerInfo.class, RemoteTrackerInfo.versionID, addr, con);
			    TrackerSystemStats tempInfo=remote.getSystemData();		    
			    addHashEntry(tem.getHostName(),tempInfo);			   
			 }
			catch (Throwable e) {
			    e.printStackTrace();					 
			 }
		}
		
	}

	public static synchronized float getCPU() {
		return CPU;
	}

	public static synchronized float getMemory() {
		return memory;
	}
}


